Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@migrations/20260410170000_execution_log.up.sql`:
- Around line 1-10: Add a foreign-key constraint tying execution_log.run_id to
the table that owns step runs/historical snapshots: alter the CREATE TABLE
execution_log to include FOREIGN KEY (run_id, dataset_id, data_product_id)
REFERENCES step_run(run_id, dataset_id, data_product_id) (or the actual run
table name/column names if different) so run_id cannot refer to a non-existent
or mismatched run; ensure the referenced columns on the run table are
unique/primary keys.

In `@migrations/20260410190000_plan_runs.down.sql`:
- Around line 1-4: The DROPs are ordered improperly: explicitly drop the index
plan_run_dataset_created_idx before dropping its parent table plan_run, and make
the down migration idempotent by using IF EXISTS on the DROP TABLE and DROP
INDEX statements for plan_run_dependency, plan_run_data_product,
plan_run_dataset_created_idx, and plan_run so the migration can be re-run
safely.

In `@migrations/20260410190000_plan_runs.up.sql`:
- Around line 15-43: Replace the single-column foreign key references on
plan_run_id with a composite foreign key (plan_run_id, dataset_id) that
references plan_run(plan_run_id, dataset_id) in both plan_run_data_product and
plan_run_dependency so rows cannot pair a plan_run_id with a different
dataset_id; specifically, update the FOREIGN KEY in plan_run_data_product
(currently FOREIGN KEY(plan_run_id) ...) to FOREIGN KEY(plan_run_id, dataset_id)
REFERENCES plan_run(plan_run_id, dataset_id) ON DELETE CASCADE, and do the same
for plan_run_dependency (replace its FOREIGN KEY(plan_run_id) ... with a
composite FK on (plan_run_id, dataset_id) referencing plan_run), leaving primary
keys and other columns unchanged.

In `@scripts/plan-dag.js`:
- Around line 81-97: Extract the hardcoded color arrays in stateColors into
named constants at the top of the file (e.g., COLORS_SUCCESS, COLORS_RUNNING,
COLORS_FAILED, COLORS_QUEUED, COLORS_DEFINED, COLORS_WAITING) and replace the
inline array returns in the stateColors function with those constants; ensure
each constant's name conveys the state and includes the same 4-color array
structure so callers of stateColors (stateColors function) keep identical
behavior while improving readability and maintainability.
- Around line 274-283: The global window event listeners for "mousemove",
"mouseup" (and the "resize" listener mentioned later) are never removed and will
leak when a widget is destroyed; fix by creating named/bound handler functions
stored on the record (e.g., record._onMouseMove, record._onMouseUp,
record._onResize) instead of anonymous functions used in
window.addEventListener, register those handlers with window.addEventListener in
the setup (so they close over dragging/startX/startY as needed), and then call
window.removeEventListener with the same stored handler references inside
destroyRecord to unregister them when the widget is torn down.
- Around line 17-18: The call to record.dag.destory() is a typo; inspect the
`@ebay/nice-dag-core` DAG instance API (or its TypeScript defs) to find the
correct cleanup method (likely destroy, dispose, teardown, or close) and replace
the incorrect destory invocation with the correct method on the DAG instance
(update the call site referencing record.dag.destory to record.dag.destroy or
the library's actual cleanup method name and ensure any required arguments or
await/async usage is handled).

In `@src/core.rs`:
- Around line 358-360: The current code path calls
plan_run_active_by_dataset_select(tx, plan.dataset.id) and returns an existing
active run, which causes explicit reruns to be attached to the old run; change
the logic so that when invoking the rerun flow you pass a reuse_active = false
(or otherwise detect an explicit rerun) and, in that case, do NOT call
plan_run_active_by_dataset_select to return the active run—instead create a
fresh plan_run record for the new execution (or alternatively reject the rerun
if you prefer) so each rerun produces a distinct historical run; update the
caller that starts full-plan reruns to pass reuse_active=false (or add a
parameter) and branch in the function that currently checks
plan_run_active_by_dataset_select(tx, plan.dataset.id) to skip returning the
existing run when reuse_active is false.
- Around line 751-758: You currently spawn a child process via command.spawn()
and immediately drop the Child handle (variable child) after logging, which
leaves exited workers unreaped; ensure the Child is awaited/reaped rather than
dropped — either call child.wait() (or .await if using tokio::process::Child) or
hand the Child to a background reaper task (e.g., tokio::spawn or
std::thread::spawn) that waits on the child and logs its exit status; locate the
command.spawn() call and the child variable in the block with the info! log and
replace the immediate drop with an explicit wait or a spawned reaper that owns
child.

In `@src/db.rs`:
- Around line 391-426: The selector execution_logs_by_dataset_select performs an
unbounded fetch_all() and must be limited; change its signature to accept
pagination parameters (e.g., limit: i64 and an optional cursor like
after_created_at: DateTime<Utc> and after_log_id: LogId or a single cursor
token), modify the SQL to add a keyset WHERE clause such as (log.created_at,
log.log_id) > ($2, $3) and a LIMIT $4 (keeping ORDER BY log.created_at,
log.log_id), and use the bounded query to return at most limit rows; apply the
same pattern to the other log selector mentioned so UI readers cannot force
unbounded in-memory loads.
- Around line 741-774: The INSERT uses ON CONFLICT DO NOTHING which returns no
row, so fetch_one() throws on duplicate inserts; update
plan_run_dependency_insert to call fetch_optional() for the INSERT and if it
returns None, issue a follow-up SELECT to load the existing PlanRunDependency
(filtering by plan_run_id, parent_id, child_id) and return that result;
reference the function plan_run_dependency_insert, the INSERT query that uses ON
CONFLICT (plan_run_id, parent_id, child_id) DO NOTHING, and the
PlanRunDependency mapping so the code returns the existing row when the INSERT
yielded no row.

In `@src/main.rs`:
- Around line 128-129: The current initialization of worker_key uses
dotenvy::var("WORKER_KEY").unwrap_or("abc123".to_string()) which silently
defaults to a known secret; change it to require the env var and fail startup
when missing by replacing the unwrap_or usage for worker_key with a fallible
check (e.g. dotenvy::var("WORKER_KEY").expect("WORKER_KEY must be set") or
propagate the Result so main returns an Err) so that the variable worker_key
cannot be set to a default value and process startup aborts if WORKER_KEY is not
provided.

In `@src/model.rs`:
- Around line 210-227: Replace the free-form String for the stream field with a
dedicated enum to restrict allowed stream values: define a LogStream (or
ExecutionStream) enum with the exact variants the worker emits, derive the
needed traits (Clone, Debug, PartialEq, Enum/GraphQL mapping,
Serialize/Deserialize) and use that enum for ExecutionLog::stream and
ExecutionLogParam::stream (and any related input/output GraphQL types). Update
all places that construct or persist ExecutionLog/ExecutionLogParam to convert
between String and the enum (or propagate the enum directly), add a DB migration
or implement/from_sql and into_sql conversions if the column stays a string, and
update tests/usages to use the enum variants instead of arbitrary strings.

In `@src/ui/page.rs`:
- Around line 379-381: The code currently injects raw dag_nodes_json using
PreEscaped in the script tag (class "fletcher-dag-data"), which allows stored
XSS if persisted fields contain "</script>"; stop using PreEscaped and instead
serialize dag_nodes_json with serde_json::to_string and then escape it before
embedding (e.g., run html-escape on the JSON or at minimum replace occurrences
of "</script>" with "<\\/script>" and escape '<', '>', '&'), or alternatively
render it as a safe text node rather than PreEscaped; update the rendering in
page.rs where PreEscaped(dag_nodes_json.to_string()) is used to output the
escaped/encoded JSON so the script block cannot be broken out of.

In `@src/worker.rs`:
- Around line 331-343: The data-product update currently injects the
worker-local filesystem path via the "link" field (see the JSON payload
constructed in the client.put call for "data_product/{}/update" which references
args.script.display()), which leaks host internals and may be rendered as a
public URL; change the payload so "link" is null (or omit the "link" key)
instead and, if you need to keep the script path for debugging, move it into
internal worker metadata rather than the public data-product update (e.g., store
under an internal worker-only field or a local log), ensuring args.script is no
longer serialized into the outbound JSON.
- Around line 99-100: stream_output() currently awaits post_log() for each line
and uses a reqwest Client with no timeouts; change it to push log lines into an
mpsc channel and spawn a dedicated background task that consumes the channel and
calls post_log() so the read/pipe draining loop never blocks on HTTP; build the
reqwest client with explicit timeouts via Client::builder().connect_timeout(...)
and timeout(...) (or equivalent) to avoid unbounded stalls when creating the
Client in Client::builder().build()? (update: set these timeouts when
constructing the client used by the background sender). Also stop persisting the
local path into the "link" field in update_state() (currently using
args.script.display().to_string()); introduce a new field (e.g., "script_path"
or "local_path") or omit the local filesystem path entirely and ensure "link"
only contains external URLs, updating the state payload construction in
update_state() accordingly so the API tests and semantics are preserved.

---

Outside diff comments:
In `@src/error.rs`:
- Around line 155-401: Add unit tests for the new Error variants (Io, Join,
Reqwest, Uuid, WorkerExit) mirroring the existing pattern: for each variant
(Error::Io, Error::Join, Error::Reqwest, Error::Uuid, Error::WorkerExit)
construct a representative error instance (e.g., std::io::Error for Io, a
JoinError for Join, a reqwest error for Reqwest, a uuid::Error for Uuid, and the
WorkerExit variant), call into_poem_error() on the Error variant, and assert
that poem_error.status() equals the expected poem::http::StatusCode as defined
by the into_poem_error mapping; add these tests to the #[cfg(test)] mod tests
alongside the other tests to ensure mappings remain correct.